package com.itcj.dmp.report

import com.itcj.dmp.utils.KuduHelper
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

object RegionReportProcess extends CommonReport {

  def main(args: Array[String]): Unit = {
    process()
  }

  override def odsTablename: String = "ODS_" + KuduHelper.today()

  override def resultTableName: String = "ODS_Count_" + KuduHelper.today()

  override def keys: List[String] = List("region","city")

  override def processDf(sourceDf: DataFrame): DataFrame = {
    sourceDf.groupBy("region","city").count()
  }

  override def schemaChange(schema: StructType)= {
    for (key <- keys) {
      val keyindex = schema.fieldIndex(key)
      schema.fields.update(keyindex,new StructField(key, StringType, false))
    }
  }
}
